package com.wolffy.flink.partition;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class PartitionDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 7777);


        // shuffle随机分区: random.nextInt(下游算子并行度)
        socketDS.shuffle().print();

        // rebalance轮询：nextChannelToSendTo = (nextChannelToSendTo + 1) % 下游算子并行度
        // 如果是 数据源倾斜的场景， source后，调用rebalance，就可以解决 数据源的 数据倾斜
        socketDS.rebalance().print();

        //rescale缩放： 实现轮询， 局部组队，比rebalance更高效
       socketDS.rescale().print();

        // broadcast 广播：  发送给下游所有的子任务
        socketDS.broadcast().print();

        // global 全局： 全部发往 第一个子任务
        // return 0;
        socketDS.global().print();

        // keyby: 按指定key去发送，相同key发往同一个子任务
        // 总结： Flink提供了 7种分区器+ 1种自定义

        env.execute();
    }
}
